package com.omega.framework.common.util;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 分页工具类
 *
 * @author Lv.
 * @date 2023/4/3 16:18
 */
public class FutureUtil {

    private static final Integer DEFAULT_PAGE_SIZE = 1000;

    /**
     * 将一批数据拆分多组，并行分别做某一个事件，同步返回结果集
     * （如一批id查询数据库，调用此方法可以设置每组500个id并行获取查询结果同步返回）
     *
     * @param ids       一批待分组的数据id
     * @param queryFunc 执行方法
     * @param pageSize  每组数据量
     * @param <T>       数据集
     * @return 结果数据
     */
    public static <T> CompletableFuture<List<T>> asyncDoSomething(List<Long> ids, Function<List<Long>, List<T>> queryFunc, Integer pageSize) {
        try {
            if (pageSize == null || pageSize <= 0) {
                pageSize = DEFAULT_PAGE_SIZE;
            }
            ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("test" + "-thread-%d").build();
            //io密集型，CPU核数/（1-阻塞系数）
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 20,
                    60L, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(100),
                    namedThreadFactory);

            // 将ID列表按照pageSize大小分组，并使用CompletableFuture.supplyAsync方法异步执行查询操作
            List<CompletableFuture<List<T>>> futures = Objects.requireNonNull(fixedGrouping(ids, pageSize))
                    .stream()
                    .map(list -> CompletableFuture.supplyAsync(() -> queryFunc.apply(list), threadPoolExecutor))
                    .collect(Collectors.toList());
            threadPoolExecutor.shutdown();
            // 使用CompletableFuture.allOf方法等待所有异步操作执行完成
            return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                    .thenApply(v -> futures.stream()
                            .map(CompletableFuture::join)
                            .flatMap(List::stream)
                            .collect(Collectors.toList()));
        } catch (Exception e) {
            return CompletableFuture.completedFuture(new ArrayList<>());
        }
    }

    /**
     * 分组数据按照数据量
     *
     * @param source   数据集
     * @param pageSize 每组数据量
     * @param <T>      数据
     * @return 每组数据
     */
    public static <T> List<List<T>> fixedGrouping(List<T> source, Integer pageSize) {

        if (CollectionUtils.isEmpty(source) || pageSize <= 0) {
            return Collections.emptyList();
        }
        int count = getCount(source.size(), pageSize);
        List<List<T>> result = new ArrayList<>();
        //分组策略，可扩展
        for (int i = 0; i < count; i++) {
            int currentGroup = i + 1;
            result.add(listGroup(source, currentGroup, pageSize));

        }
        return result;
    }

    /**
     * 根据长度及每组需要的条数，获取分组数
     *
     * @param size     数据长度
     * @param pageSize 每组需要条数
     * @return 分组数
     */
    private static int getCount(int size, Integer pageSize) {
        // 页数
        int count;
        if (size % pageSize == 0) {
            count = size / pageSize;
        } else {
            count = size / pageSize + 1;
        }
        return count;
    }

    /**
     * 获取每组数据
     *
     * @param list         总数据
     * @param currentGroup 当前组标识
     * @param pageSize     每组数据量
     * @param <T>          数据
     * @return 每组数据
     */
    public static <T> List<T> listGroup(List<T> list, int currentGroup, int pageSize) {
        if (CollectionUtils.isEmpty(list)) {
            return Collections.emptyList();
        }

        // 记录总数
        int count = list.size();
        // 分组数
        int groupCount = getCount(count, pageSize);
        // 开始索引
        int fromIndex;
        // 结束索引
        int toIndex;
        if (currentGroup != groupCount) {
            fromIndex = (currentGroup - 1) * pageSize;
            toIndex = fromIndex + pageSize;
        } else {
            fromIndex = (currentGroup - 1) * pageSize;
            toIndex = count;
        }
        return list.subList(fromIndex, toIndex);
    }


    public static void main(String[] args) throws ExecutionException, InterruptedException {
        List<Long> ids = new ArrayList<>();
        ids.add(12313123123L);
        ids.add(12313123122L);
        ids.add(12313123121L);
        ids.add(12313123120L);
        ids.add(12313123129L);

        ids.add(12313123128L);
        ids.add(12313123127L);
        ids.add(12313123126L);
        ids.add(12313123125L);

        ids.add(12313123124L);
        ids.add(12313123223L);
        ids.add(12313123323L);
        ids.add(12313123423L);
        CompletableFuture<List<Object>> future = asyncDoSomething(ids, FutureUtil::getObject, 5);
        List<Object> objects = future.get();
        for (int i = 0; i < objects.size(); i++) {
            System.out.println(objects.get(i));
        }
    }


    private static List<Object> getObject(List<Long> ids) {
        System.out.println("当前线程名称：" + Thread.currentThread().getName());
        System.out.println("当前时间戳：" + System.currentTimeMillis());
        Map<Long, Object> maps = new HashMap<>();
        maps.put(12313123123L, "第一条数据");
        maps.put(12313123122L, "第二条数据");
        maps.put(12313123121L, "第三条数据");
        maps.put(12313123120L, "第四条数据");
        maps.put(12313123129L, "第五条数据");

        maps.put(12313123128L, "第六条数据");
        maps.put(12313123127L, "第七条数据");
        maps.put(12313123126L, "第八条数据");
        maps.put(12313123125L, "第九条数据");

        maps.put(12313123124L, "第十条数据");
        maps.put(12313123223L, "第十一条数据");
        maps.put(12313123323L, "第十二条数据");
        maps.put(12313123423L, "第十三条数据");
        List<Object> objects = new ArrayList<>();
        for (Long id : ids) {
            for (Map.Entry<Long, Object> longObjectEntry : maps.entrySet()) {
                if (longObjectEntry.getKey().equals(id)) {
                    objects.add(longObjectEntry.getValue());
                }
            }
        }
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return objects;
    }
}
